package cn.com.captureNetPacket.voice;

import java.io.File;
import java.io.UnsupportedEncodingException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import cn.com.captureNetPacket.model.*;
import org.jnetpcap.protocol.voip.Rtp;

import cn.com.captureNetPacket.Analysis;
import cn.com.captureNetPacket.CaptureNetPacketService;
import cn.com.captureNetPacket.record.WavFileInfo;
import cn.com.captureNetPacket.util.RedisConfig;
import cn.com.captureNetPacket.util.Tools;
import cn.com.captureNetPacket.util.WavFileUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Slf4j
@Data
public class AnalysisRTP extends Analysis {

//	RTP 报文解释，一下可获取的源码提供，源码有图和位置信息等，一看就明白。源码使用的是，里面的话。  RFC 3550 RTP July 2003 Note。后面是获取的方法名
//	V：RTP协议的版本号，占2位，当前协议版本号为2。    Rtp.version();可获取
//	P：填充标志，占1位，如果P=1，则在该报文的尾部填充一个或多个额外的八位组，它们不是有效载荷的一部分。Rtp.hasPadding() 可判断。  Rtp.paddingLength()可知道 额外填充的部分。应该是8的倍数，取出后可转换。
//	X：扩展标志，占1位，如果X=1，则在RTP报头后跟有一个扩展报头。  Rtp.hasExtension();  可取到。取到后可进行处理。展开解释：X 扩展位 根据RFC5285的规定，RTP扩展头部有两种格式：one byte header和two byte header。参考这里有提到扩展X的意义： https://www.jianshu.com/p/efc5ef2113da
//	 CC：CSRC计数器，占4位，指示CSRC 标识符的个数。Rtp.count();   已经实现 Rtp.csrc()调用了这个参数，直接取出csrc了。后面讲
//	M: 标记，占1位，不同的有效载荷有不同的含义，对于视频，标记一帧的结束；对于音频，标记会话的开始。 Rtp.hasMarker();
//	PT: 有效载荷类型，占7位，用于说明RTP报文中有效载荷的类型，如GSM音频、JPEM图像等,在流媒体中大部分是用来区分音频流和视频流的，这样便于客户端进行解析。Rtp.type();获取枚举Rtp.typeEnum();
//	序列号：占16位，用于标识发送者所发送的RTP报文的序列号，每发送一个报文，序列号增1。这个字段当下层的承载协议用UDP的时候，网络状况不好的时候可以用来检查丢包。
//	同时出现网络抖动的情况可以用来对数据进行重新排序，在helix服务器中这个字段是从0开始的，同时音频包和视频包的sequence是分别记数的。Rtp.sequence();可获取。
//	 时戳(Timestamp)：占32位，时戳反映了该RTP报文的第一个八位组的采样时刻。接收者使用时戳来计算延迟和延迟抖动，并进行同步控制。Rtp.timestamp();可获取
//	同步信源(SSRC)标识符：占32位，用于标识同步信源。该标识符是随机选择的，参加同一视频会议的两个同步信源不能有相同的SSRC。  Rtp.ssrc();可获取。
//	特约信源(CSRC)标识符：每个CSRC标识符占32位，可以有0～15个。每个CSRC标识了包含在该RTP报文有效载荷中的所有特约信源。Rtp.csrc();源码调用了Rtp.count()方法来取csrc。
	public static HashMap<String ,AnalysisRTP> analysisRTPs=new  HashMap<String,AnalysisRTP>();
	/**
	 * 正方向 rtp 缓存数组
	 */
	LinkedList<RtpDto> positiveRtpsTemp = new LinkedList();


	/**
	 * 反方向 rtp 缓存数组
	 */
	LinkedList<RtpDto> reverseRtpsTemp = new LinkedList();

	private Thread iniRtpInfoThread;
	private boolean iniRtpInfoThreadStoped=false;
	private    ConcurrentHashMap<Integer ,RtpDto> positiveRtps =new ConcurrentHashMap<Integer,RtpDto>();
	private    ConcurrentHashMap<Integer ,RtpDto> reverseRtps=new ConcurrentHashMap<Integer,RtpDto>();
	// 被唤醒的线程去处理相关任务信息，解析信令和发送流
	 VoiceInfoPush voiceInfoPush;
	// 绑定信令
	private SignallingInfo signallingInfo;
	private String positivePath =null,reversePath=null,saveFilePath=null;
//	接受到的最后的 TimeTamp


	//	接受到的最后的 rtp包
	RtpDto positiveReciveLastRtpDto =null, reverseReciveLastRtpDto =null;

//   最后接受到RTP包的时间
	long positiveReciveLastTime =0,reverseReciveLastTime=0;


	LocalDateTime initTime=LocalDateTime.now();
	//  记录上一次接受的的TimestampSeq 避免添加rtp包的时候，需要坚持时再次计算TimestampSeq
	int positiveReciveLastTimestampSeq =0,reverseReciveLastTimestampSeq=0;




	private WavFileInfo wavFileInfo=null;
	DeviceInfo deviceInfo=null;
	//客户端是否为先发起的包
	private boolean clientToServerIsInPositiveDirection =false;





	// 文件缓存buffer
	byte[] positiveFileCacheBytes,  reverseFileCacheBytes;
	 private static AtomicInteger  channelCount=new AtomicInteger(0);

	static final Logger rtpPackagesLog = LoggerFactory.getLogger("rtpPackagesLog");

	public static void addRtp(NetworkInfo netInfo, Rtp rtp){

		if(!isNeedPacket(netInfo)) {
			return;
		}
		AnalysisRTP analysisRTP= getAnalysisRTP(netInfo);
		if(analysisRTP==null){
			analysisRTP= getAnalysisRTP(netInfo);
              synchronized (analysisRTPs) {
				  //没有就初始化，开启线程去检测
				  analysisRTP = new AnalysisRTP(netInfo);
				  analysisRTP.setPriority(NORM_PRIORITY);
				  analysisRTP.setName("AnalysisRTP-Main"+netInfo.getNetworkInfoKey());
				  if(getAnalysisRTP(netInfo)==null) {
					  analysisRTPs.put(netInfo.getNetworkInfoKey(), analysisRTP);
//			analysisRTP
					  CaptureNetPacketService.getTaskService().execute(analysisRTP);
					  log.debug("flow 初始化语音流分析程序结束 analysisRTP。 sourceIP:{} port: {} mac:{}->destinationIP:{}  port: {}  mac: {} timestamp: {} sequence: {} networkInfoKey: {}", netInfo.getSourceIP(), netInfo.getSourcePort(), netInfo.getSourceMac(), netInfo.getDestinationIP(), netInfo.getDestinationPort(), netInfo.getDestinationMac(), rtp.timestamp(), rtp.sequence(), netInfo.getNetworkInfoKey());
				  }else {
					  analysisRTP= getAnalysisRTP(netInfo);
				  }
				  }
			  }
		if(analysisRTP.isStopThread()){
			return;
		}
		long curentTime=System.currentTimeMillis();
		RtpDto recRtpDto=RtpDto.builder().reciveTime(curentTime).build();
		recRtpDto.original2Dto(rtp);
       if(analysisRTP.positiveNetworkInfo.getNetworkInfoKey().equals(netInfo.getNetworkInfoKey())){

		   analysisRTP.positiveRtpsTemp.addLast(recRtpDto);

			   analysisRTP.positiveReciveLastRtpDto = recRtpDto;
		   analysisRTP.positiveReciveLastTime =curentTime;

       }else{
		   if(CaptureNetPacketService.getAppConfig().isRemoveFront()){
			   // 如果需要移除前面的振铃包，则把方向前面的包清理掉。
			   analysisRTP.positiveRtpsTemp.clear();
			   analysisRTP.positiveRtpsTemp=new LinkedList<>();
		   }else{
			   if(analysisRTP.reverseReciveLastRtpDto== null) {
				   // 如果需要不移除前面的振铃的包， 反方向第一次没有接受到包，需要补充包对齐。
				   analysisRTP.reverseRtpsTemp.addAll(analysisRTP.positiveRtpsTemp);
				   rtpPackagesLog.debug("addRtp 将正方向的数据加到反方向补充:  sourceIP:{} port: {} mac:{}->destinationIP:{}  port: {}  mac: {} timestamp: {} sequence: {}  缓存语音包数量: 正向: {},反向: {}",netInfo.getSourceIP(), netInfo.getSourcePort() , netInfo.getSourceMac(),netInfo.getDestinationIP(), netInfo.getDestinationPort(),netInfo.getDestinationMac(),rtp.timestamp(),rtp.sequence() ,analysisRTP.positiveRtpsTemp.size(),analysisRTP.reverseRtpsTemp.size());
			   }
		   }
		   analysisRTP.reverseRtpsTemp.addLast(recRtpDto);
			   analysisRTP.reverseReciveLastRtpDto = recRtpDto;
		   analysisRTP.reverseReciveLastTime=curentTime;
       }
       analysisRTP.lastGetPacketDateTime=  Instant.ofEpochMilli(curentTime).atZone(ZoneId.systemDefault()).toLocalDateTime();
		rtpPackagesLog.debug("addRtp 原始数据添加成功:  sourceIP:{} port: {} mac:{}->destinationIP:{}  port: {}  mac: {} timestamp: {} sequence: {}  缓存语音包数量: 正向: {},反向: {}",netInfo.getSourceIP(), netInfo.getSourcePort() , netInfo.getSourceMac(),netInfo.getDestinationIP(), netInfo.getDestinationPort(),netInfo.getDestinationMac(),rtp.timestamp(),rtp.sequence() ,analysisRTP.positiveRtpsTemp.size(),analysisRTP.reverseRtpsTemp.size());
		if(analysisRTP.getPositiveRtpsTemp().size()>CaptureNetPacketService.getAppConfig().getRtpBatchPackagesSave()+analysisRTP.surplus ||
			   analysisRTP.getReverseRtpsTemp().size()>CaptureNetPacketService.getAppConfig().getRtpBatchPackagesSave()+analysisRTP.surplus
	   )
	   {
		   analysisRTP.awaken();
	   }
	}
	
	/**
	 *  	是否为需要抓取的分机
	 * @param netInfo
	 * @return
	 */
	public static boolean isNeedPacket(NetworkInfo netInfo) {
		if(CaptureNetPacketService.getAppConfig().isCapAllRtpIP()) {
			return true;
		}
		if(CaptureNetPacketService.getDeviceInfos().containsKey(netInfo.getSourceIP())) {
			return true;
			}
		if(CaptureNetPacketService.getDeviceInfos().containsKey(netInfo.getDestinationIP()))
			{return true;
			}
		if(CaptureNetPacketService.getDeviceInfos().containsKey(netInfo.getSourceMac())) {
			return true;
		}
		if(CaptureNetPacketService.getDeviceInfos().containsKey(netInfo.getDestinationMac())) {
			return true;
		}
		return false;
	}
	
	public AnalysisRTP(NetworkInfo networkInfo){
		log.debug("AnalysisRTP 初始化了。");
		//未初始化
          super.positiveNetworkInfo =networkInfo;
          super.reverseNetworkInfo=new NetworkInfo(networkInfo.getDestinationIP(), networkInfo.getSourceIP(), networkInfo.getDestinationMac(),networkInfo.getSourceMac(), networkInfo.getTransportLayer(), networkInfo.getDestinationPort(),networkInfo.getSourcePort());
	}
	
	
	public static AnalysisRTP getAnalysisRTP(NetworkInfo networkInfo){
		// 一通会话 IP和端口只会是一个，改变了肯定是新的通话  CSRC同步源，每一个通话方向的语音流CSRC不一样。
		AnalysisRTP analysisRTP=null;
		analysisRTP=analysisRTPs.get(networkInfo.getNetworkInfoKey());
		if(analysisRTP!=null)
			return analysisRTP;
		analysisRTP=analysisRTPs.get(networkInfo.getReverseNetworkInfoKey());
		return analysisRTP;
	}

	/**
	 * 将数据包根据协议及时在缓存区进行排序，解决网络抖动出现的包乱序。
	 */
	public void analysisSortPackageData(){

		if(this.surplus>0) {
			Duration dur = Duration.between(initTime, LocalDateTime.now());
			if (this.reverseReciveLastTimestampSeq < 51 && !CaptureNetPacketService.getAppConfig().isRemoveFront() && dur.toMillis() < CaptureNetPacketService.getFluctuationRange() && !isTimeOut) {
				log.debug("analysisSortPackageData ### 。不移除振铃部分的语音包，反向的语音包，需等待(FluctuationRang){} 毫秒的数据包量或等待振铃的包接受完成后，后开始整理。---> sourceIP:{} port:{} mac:{}->destinationIP:{}  port:{}  mac:{}  positiveRtps.size: {}  reverseRtps.size: {}", CaptureNetPacketService.getFluctuationRange(), positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort(), positiveNetworkInfo.getSourceMac(), positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(), positiveNetworkInfo.getDestinationMac(), positiveRtps.size(), reverseRtps.size());
				return;
			}
		}
		analysisSortPackageData(false,reverseRtpsTemp);
		analysisSortPackageData(true,positiveRtpsTemp);
	}

	/**
	 * 重新根据Timestamp 结合接受时间进行排序
	 * @param isHandelPositive 处理正还是翻
	 * @param rtpsTemp 缓存数据区。
	 */
	public void analysisSortPackageData(boolean isHandelPositive,LinkedList<RtpDto> rtpsTemp){
		List<RtpDto> rtosSortbyTm = new ArrayList<>();
		try {
			while (rtpsTemp.size() > surplus) {
				// 移除 rtpsfactory 的第一个并返回，并添加到 需要处理的数据 。rtosSortbyTm
				rtosSortbyTm.add(rtpsTemp.poll());
			}
			List<List<RtpDto>>  groupRtpDtos= analysisSortByGroupPackageData(rtosSortbyTm);
			if(groupRtpDtos!=null && groupRtpDtos.size()>0){
				for(int i=0;i<groupRtpDtos.size();i++){
					List<RtpDto>	rtpDtos=groupRtpDtos.get(i);
					if(rtpDtos.size()>1) {
						rtpPackagesLog.debug("是否正方向: {}  . 第{}组.第一个信息:接受的时间:{} tm: {} seq: {} , 最后一个信息:接受的时间:{} tm: {} seq: {}", isHandelPositive, i, rtpDtos.get(0).getReciveTime(), rtpDtos.get(0).getTimestamp(), rtpDtos.get(0).getSequence(), rtpDtos.get(rtpDtos.size() - 1).getReciveTime(), rtpDtos.get(rtpDtos.size() - 1).getTimestamp(), rtpDtos.get(rtpDtos.size() - 1).getSequence());
					}else if(rtpDtos.size()==1){
						rtpPackagesLog.debug("是否正方向: {}  本组只有一个包. 第{}组.第一个信息:接受的时间:{} tm: {} seq: {} ", isHandelPositive, i, rtpDtos.get(0).getReciveTime(), rtpDtos.get(0).getTimestamp(), rtpDtos.get(0).getSequence());
					}
					for(int j=0;j<rtpDtos.size();j++) {
						int step = 1;
						if (j > 0 && j < rtpDtos.size()) {
							step = (int) ((rtpDtos.get(j).getTimestamp() - rtpDtos.get(j - 1).getTimestamp()) / CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize());
						}
						if (step > 0) {
							// step 等于0表示与上一个重复，每组第一个包 step均为1，后续与前一个的差值为步长。
							save2FormalRtpData(isHandelPositive, rtpDtos.get(j), step);
						}
					}

				}
			}

		}catch (Exception e){
			log.error("排序失败：isHandelPositive: {} 数据包: {}",isHandelPositive,rtosSortbyTm,e);
		}
	}

	/**
	 * 保存到正式的rtp包数据区,等待写入文件的处理。
	 * @param isHandelPositive
	 * @param rtpDto
	 */
    public void save2FormalRtpData(boolean isHandelPositive,RtpDto rtpDto,int step){
		if(isHandelPositive){
			positiveReciveLastTimestampSeq=positiveReciveLastTimestampSeq+step;
			rtpDto.setRealSeq(positiveReciveLastTimestampSeq);
			positiveRtps.put(positiveReciveLastTimestampSeq,rtpDto);
			rtpPackagesLog.debug("save2FormalRtpData 排序后的数据，接受时间： {} 正向自定义序号添加成功: {} sourceIP:{} port: {} mac:{}->destinationIP:{}  port: {}  mac: {} timestamp: {} sequence: {}  语音包数量: 正向: {},反向: {}",rtpDto.getReciveTime(),positiveReciveLastTimestampSeq,positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort() , positiveNetworkInfo.getSourceMac(),positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(),positiveNetworkInfo.getDestinationMac(),rtpDto.getTimestamp(),rtpDto.getSequence() ,positiveRtps.size(),reverseRtps.size());
		}else {
			reverseReciveLastTimestampSeq=reverseReciveLastTimestampSeq+step;
			rtpDto.setRealSeq(reverseReciveLastTimestampSeq);
			reverseRtps.put(reverseReciveLastTimestampSeq,rtpDto);
			rtpPackagesLog.debug("save2FormalRtpData 排序后的数据，接受时间： {} 反向自定义序号添加成功: {} sourceIP:{} port: {} mac:{}->destinationIP:{}  port: {}  mac: {} timestamp: {} sequence: {}  语音包数量: 正向: {},反向: {}",rtpDto.getReciveTime(),reverseReciveLastTimestampSeq,positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(),positiveNetworkInfo.getDestinationMac(),positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort() , positiveNetworkInfo.getSourceMac(),rtpDto.getTimestamp(),rtpDto.getSequence() ,positiveRtps.size(),reverseRtps.size());
		}
		rtpDto.setSave2Formal(true);
	}

	/**
	 * 根据Tm分组和再根据每组第一个接受时间排序
	 * @param sourceRtpRtos  原始乱序的数据
	 * @return 分组重组后的数据
	 */
	public  List<List<RtpDto>> analysisSortByGroupPackageData(List<RtpDto> sourceRtpRtos){
		List<List<RtpDto>> groupRtpDtos = new ArrayList<>();
		// 按 rto.getRtp().timestamp()  排序
		List<RtpDto> sortByTm = sourceRtpRtos.stream().sorted(Comparator.comparing(rto -> rto.getTimestamp())).collect(Collectors.toList());
		try {
			List<Integer> groupPoints = new ArrayList<>();
			for (int i = 0; i < sortByTm.size() - 1; i++) {
				//  如果中间包大于网络抖动容忍的包数量，说明要分一个组.   没两个包的 timestamp 差为 getRtpAudioPacketSize，每个包之间的时间为10ms或20ms,一般为20ms
				if (Math.abs(sortByTm.get(i + 1).getTimestamp() - sortByTm.get(i).getTimestamp()) / CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize()  > CaptureNetPacketService.getFluctuationRangeMaxLosePackets()) {
					//  出现分界点的序号。
					groupPoints.add(i+1);
				}
			}
			rtpPackagesLog.debug("analysisSortByGroupPackageData groupPoints: {}",groupPoints);
			if (groupPoints.size() > 0) {
				for (int i = 0; i < groupPoints.size() + 1; i++) {
					try {
						if (i == 0) {
							groupRtpDtos.add(sortByTm.subList(0, groupPoints.get(i)));
						} else if (i > 0 && i < groupPoints.size()) {
							groupRtpDtos.add(sortByTm.subList(groupPoints.get(i - 1), groupPoints.get(i)));
						} else {
							groupRtpDtos.add(sortByTm.subList(groupPoints.get(i - 1), sortByTm.size()));
						}
						rtpPackagesLog.debug("####正在进行分组 . 第{}组.第一个信息:接受的时间:{} tm: {} seq: {} , 最后一个信息:接受的时间:{} tm: {} seq: {}",i,groupRtpDtos.get(i).get(0).getReciveTime(),groupRtpDtos.get(i).get(0).getTimestamp(),groupRtpDtos.get(i).get(0).getSequence(),groupRtpDtos.get(i).get(groupRtpDtos.get(i).size()-1).getReciveTime(),groupRtpDtos.get(i).get(groupRtpDtos.get(i).size()-1).getTimestamp(),groupRtpDtos.get(i).get(groupRtpDtos.get(i).size()-1).getSequence());
					} catch (Exception e) {
						log.warn("重新分组时失败 groupPoints（分组点）", groupPoints, sortByTm,e);
					}
				}
				// 根据每组第一个的最早的接受时间进行排序后合并
				groupRtpDtos = groupRtpDtos.stream().sorted(Comparator.comparing(groupRtpDto -> groupRtpDto.get(0).getReciveTime())).collect(Collectors.toList());
			}
		}catch (Exception e){
			log.warn("重新分组时失败 排序后的数据 {}", sortByTm,e);

		}
		if(groupRtpDtos.size()<1){
			groupRtpDtos.add(sortByTm);
		}
		return groupRtpDtos;
	}



	public void AnalysisRTPData(){

		 try {
			 Integer[] keys =null,reverseKeys=null;
				 
				 byte[] realTimeData = null,reverseRealTimeData= null;
		 //如果数据包超过一定的数量需要将数据写入文件，并留下10个左右的包，超时的时候全部处理，避免有包还没到的情况出现排序跳动太大的情况。下一次处理前面的超时的包。
		 // 处理没超时的包
				 try {
						 if(positiveRtps.size()>CaptureNetPacketService.getAppConfig().getRtpBatchPackagesSave()+surplus ||reverseRtps.size()>CaptureNetPacketService.getAppConfig().getRtpBatchPackagesSave() +surplus || isTimeOut) {
							 log.debug("AnalysisRTPData ### ---> sourceIP:{} port:{} mac:{}->destinationIP:{}  port:{}  mac:{}  positiveRtps.size: {}  reverseRtps.size: {}", positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort() , positiveNetworkInfo.getSourceMac(), positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(), positiveNetworkInfo.getDestinationMac(), positiveRtps.size(),reverseRtps.size());

							 packageRtpDataSaveCache(positiveRtps, positivePath, positiveFileCacheBytes,isTimeOut);
							 packageRtpDataSaveCache(reverseRtps, reversePath, reverseFileCacheBytes,isTimeOut);
			 		} 
				 }catch (Exception e) {
					 log.error("AnalysisRTPData save Batch data ---> sourceIP:{} port:{} mac:{}->destinationIP:{}  port:{}  mac:{}   Error :", positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort() , positiveNetworkInfo.getSourceMac(), positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(), positiveNetworkInfo.getDestinationMac(),e);
				}
			 if(CaptureNetPacketService.getAppConfig().isRealTimeTransmissionData() && keys!=null &&reverseKeys!=null) {
				 try {
					 log.debug(" add voiceInfoPush ---> sourceIP:{} port:{} mac:{}->destinationIP:{}  port:{}  mac:{}  positiveRtps.size: {}  reverseRtps.size: {}", positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort() , positiveNetworkInfo.getSourceMac(), positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(), positiveNetworkInfo.getDestinationMac(), positiveRtps.size(),reverseRtps.size());
					 // 进入发送实时语音流的逻辑
					 //    if(realTimeData!=null) 上面处理了，有数据才会继续。否则不会进实时发送语音流。                       是否发送受上面的逻辑影响,如果存文件的保留过大，不会走到这一步，也会影响发送实时的速率
							 if( voiceInfoPush!= null ) 
							 {
								 log.debug(" need voiceInfoPush --->  DeviceNumber: {} SourceIP: {} DestinationIP: {} positiveRtps: {} reverseRtps: {}",voiceInfoPush.getSignallingInfo().getDeviceNumber(), positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getDestinationIP(), positiveRtps.size(),reverseRtps.size());
									
								 if(realTimeData!=null || reverseRealTimeData!=null )
							 {
								 // 保证客户端发送的语音流始终在同一声道
							if(!clientToServerIsInPositiveDirection && !CaptureNetPacketService.getAppConfig().isRealTimeCustomerPositive()) {
								 if(realTimeData!=null)
							 voiceInfoPush.setPositiveRealTimeData(realTimeData);
			//				 voiceInfoPush.setReverseRealTimeData(reverseRealTimeData);
							 }else {
								 if(reverseRealTimeData!=null)
								 voiceInfoPush.setPositiveRealTimeData(reverseRealTimeData);
			//					 voiceInfoPush.setReverseRealTimeData(realTimeData);
							 }
							 log.debug("pre  voiceInfoPush ---> sourceIP:{} port:{} mac:{}->destinationIP:{}  port:{}  mac:{}  positiveRtps.size: {}  reverseRtps.size: {}", positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort() , positiveNetworkInfo.getSourceMac(), positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(), positiveNetworkInfo.getDestinationMac(), positiveRtps.size(),reverseRtps.size());
								
							// 语音流处理完成，发送到推送程序完成推送并识别处理逻辑.设置音频格式
							voiceInfoPush.appendVoiceData();
							 if(voiceInfoPush.getWavFileInfo()==null) 
								 voiceInfoPush.setWavFileInfo(wavFileInfo);
							 voiceInfoPush.interrupt();
							 log.debug("end  voiceInfoPush ---> sourceIP:{} port:{} mac:{}->destinationIP:{}  port:{}  mac:{}  positiveRtps.size: {}  reverseRtps.size: {}", positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort() , positiveNetworkInfo.getSourceMac(), positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(), positiveNetworkInfo.getDestinationMac(), positiveRtps.size(),reverseRtps.size());
								
							 
							 }
							 }
				 }catch (Exception e) {
					 log.error("AnalysisRTPData voiceInfoPush ---> sourceIP:{} port:{} mac:{}->destinationIP:{}  port:{}  mac:{}   Error :", positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort() , positiveNetworkInfo.getSourceMac(), positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(), positiveNetworkInfo.getDestinationMac(),e);
				}
			 }
		 } catch (Exception e) {
			 log.error("AnalysisRTPData AnalysisRTPData ---> sourceIP:{} port:{} mac:{}->destinationIP:{}  port:{}  mac:{}   Error :", positiveNetworkInfo.getSourceIP(), positiveNetworkInfo.getSourcePort() , positiveNetworkInfo.getSourceMac(), positiveNetworkInfo.getDestinationIP(), positiveNetworkInfo.getDestinationPort(), positiveNetworkInfo.getDestinationMac(),e);
			}
	}
	
	/**
	 * 
		* <strong>Title: <strong>
		* <p>Description: 将数据写入缓存。<p>
		* <p>Company: </p> 
		*<strong>Copyright:</strong> Copyright (c) 2019
		* @version: 1.0
		* @author  ASUS:augustWei
		* @date  2020 下午11:17:43
		* @param rtps 所有的RTP包
		* @param pathtemp 临时文件存储的路径
		* @param fileCacheBytes  文件存储的临时数组
		* @param forceWriteCache  是否全部写入缓存
		* @return
	 */
	public byte[] packageRtpDataSaveCache(ConcurrentHashMap<Integer ,RtpDto> rtps,String pathtemp,byte[] fileCacheBytes,boolean forceWriteCache) {

		// 将key排序，避免 超过 Short.MAX_VALUE. 这里就进行排序是为了尽量减少，后面处理时，正向解析比较慢的情况，在正向完成后，反向又增加了不少的数据包，导致语音通道数据错位的情况。这样做尽量降低了错位的情况。
         if(rtps ==null ) return null;
		// 处理seq可能越界的情况。
//		keys 排序后的包顺序，并且不存在数组越界的情况，数组越界的情况在handelOriginBatch函数中2次处理。
		Integer[] keys=rtps.keySet().toArray(new Integer[rtps.keySet().size()]);
		 if(keys == null || keys.length<1 )return null;

		boolean isHandelPositiveRtps = positivePath.equals(pathtemp);

		 // 每次留是个包保证下次排序，最后10个都到了还有没到的，说明最后10之前的有丢包，直接处理掉，正常情况下最后10个之前的应该是按顺序的。最终超时后对所有剩余的包进行处理。
		int lastPacketNumber=keys.length<2?1:keys.length-1;

		 int step=0,bufferSize=0,bufferSeq=keys[lastPacketNumber]-keys[0]+1;RtpDto rtp1=null;


		bufferSize=(bufferSeq)*(CaptureNetPacketService.getAppConfig().getOtherRtpAudioPacketSize()>0?CaptureNetPacketService.getAppConfig().getOtherRtpAudioPacketSize():CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize());
		 
//		 byte[] temp=new byte[(keys.length-surplus)*214],factData = null;
		 
		 byte[] temp=null,factData = null;
		 log.debug("整理开始，packageRtpDataSaveCache : 总共{} 个数据包,本次处理: {},第一包序号: {}   最后一包序号 : {} 到缓存，bufferSize: {} 本轮是否强制写入缓存路径: {} 缓存临时路径: {}",rtps.size(),keys.length,keys[0],keys[lastPacketNumber],bufferSize,forceWriteCache,pathtemp);
		 temp=new byte[bufferSize];
		 // 复制到的数据的位置
		 int copyPos=0;
		 try {
		 for (int i=0;i< keys.length;i++) {
			 int currentSeq=keys[i],nextSeq=0;
			 rtp1=rtps.get(currentSeq);
			 if(rtp1==null){
			 	// 垃圾序号，可能已经被添加了。

				 // 清理掉前面的垃圾数据
				 removeOldSeq(rtps,currentSeq);
				 //移除数据
				 rtps.remove(currentSeq);
				 log.debug("packageRtpDataSaveCache : 清理空包并删除{}，剩余 rtp包 {}个数据.",currentSeq,rtps.size());
			 	continue;
			 }
			 
			 if(i<keys.length-1) {
				 nextSeq=keys[i+1];
			 step=nextSeq-currentSeq;
			 }
			 

			 // step  =0 表示是本轮的第一个语音包,step=1表示是连续的包，step  0,1都是有效的语音数据

			   if(step>=CaptureNetPacketService.getFluctuationRangeMaxLosePackets()){
				 //  说明前面超过大量包丢掉了,本轮无需累加加数据了。也有可能是前一轮的垃圾数据导致的最小值不对了。
				 log.debug("packageRtpData: lose large 出现大量丢包,丢了{}个包，不做静音数据填充，当前序号:{}, 随后序号: {} ,--------> 补包后的位置: {} 临时数组大小: {}",step-1,currentSeq,nextSeq,copyPos,bufferSize);
			 }
			 
			 
			 //----------------***********************顺序正常需要识别出音频信息
			 // step如果大于1说明掉包了。不能用来计算速率
			 if(wavFileInfo==null && step==1) {
				 long tempSimple =rtps.get(nextSeq).getTimestamp()- rtp1.getTimestamp();
//				 0
//				 PCM μ -Law
//				 8
//				 PCM-A Law
//				 9
//				 G..722 audio codec
//				 4
//				 G..723 audio codec
//				 15
//				 G..728 audio codec
//				 18
//				 G..729 audio codec
//				 34
//				 G..763 audio codec
//				 31
//				 G..761 audio codec
				 short audioFormat=6;
				 if(rtp1.getType()==0) {
					 audioFormat=7;
				 }
				 creatWavFileInfo(tempSimple,audioFormat);
			 }
			 //----------------***********************
			 
			 try {
			 // 复制数据
//			 rtp1.getPayload();
				 if(rtp1.getPayload()!=null) {
					 if (rtp1.getPayload().length == CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize() ) {
//			 System.arraycopy(rtp1.getPayload(),0,temp,copyPos,rtp1.getPayload().length);
//			 copyPos=copyPos+rtp1.getPayload().length;
						 System.arraycopy(rtp1.getPayload(), 0, temp, copyPos,CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize());
						 copyPos = copyPos + CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize();
					 }else if (rtp1.getPayload().length != CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize()) {
						 try {
						 	if(CaptureNetPacketService.getAppConfig().isAcceptOtherRtpPacketSize()){
						 		if(CaptureNetPacketService.getAppConfig().isAcceptOtherRtpAudioPacketSize()) {
						 			// 接受audioPacketSize 大小的数据，只留有效数据的最后一部分数据
									System.arraycopy(rtp1.getPayload(), rtp1.getPayload().length - CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize(), temp, copyPos,CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize());
									copyPos = copyPos + CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize();
								}else {
						 			// 接受所有的有效数据
									System.arraycopy(rtp1.getPayload(), 0, temp, copyPos, rtp1.getPayload().length);
									copyPos = copyPos + rtp1.getPayload().length;
								}
							}
							 // rtp包头为12字节
							 //  rtp1.getPacket().size() 190   rtp1.getPayload().length  136  rtp1.getHeader().length 12  rtp1.getLength()  12
							 //  rtp1.getPacket().size() 169   rtp1.getPayload().length  115  rtp1.getHeader().length 12  rtp1.getLength()  12
							 //  rtp1.getPacket().size() 60   rtp1.getPayload().length  4  rtp1.getHeader().length 12  rtp1.getLength()  12

                               if(1==2) {
                                    //确认异常包是否为心跳包，心跳涉及补报的逻辑
								   System.arraycopy(rtp1.getPayload(), 0, temp, copyPos, CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize());
								   copyPos = copyPos + CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize();
							   }
//							 log.debug("packageRtpData--------->>>>: 出现 包大小异常的包  当前序号:{}, 随后序号: {}  当前缓冲区的位置: {} 临时数组大小: {}, 出现 Payload.size: {} 不为 audioPacketSize的数据, 头长度: {} rtp1.getLength: {} getPacketSize: {}", currentSeq, nextSeq, copyPos, bufferSize, rtp1.getPayload().length, rtp1.getHeader().length, rtp1.getLength(), rtp1.getPacket().size());
							 log.debug("packageRtpData--------->>>>: 出现 包大小异常的包  当前序号:{}, 随后序号: {}  当前缓冲区的位置: {} 临时数组大小: {}, 出现 Payload.size: {} 不为 audioPacketSize的数据, 头长度: {} rtp1.getLength: {} getPacketSize: {}", currentSeq, nextSeq, copyPos, bufferSize, rtp1.getPayload().length);
//				 log.debug("currentSeq {}-------------->>>typeEnum： {}",currentSeq,rtp1.typeEnum());

						 } catch (Exception e) {
							 log.error("packageRtpData currentSeq :{} Exception: {}", currentSeq, e.getMessage());
						 }
					 }
				 }

//  这里使用上一个数据包往后填充，所以判断下上一个是否为空。如果使用静音补包，可去掉此条件。
				 if(rtp1.getPayload()!=null) {
					 if(i<keys.length-1){
					 //丢包少于60秒可以补,一秒钟50个包，
					 if (step > CaptureNetPacketService.getAppConfig().getAllowSkipPacketNum()+1 && step < CaptureNetPacketService.getFluctuationRangeMaxLosePackets()) {
						 log.debug("packageRtpData: 出现丢包 ,当前位置需要补{}个包，当前序号:{}, 随后序号: {} ,补包位置: {} 临时数组大小: {} 序号: {}", step - 1, currentSeq, nextSeq, copyPos, bufferSize, i);
						 for (int k = 0; k < step - 1; k++) {
							 // 补位后继续下一轮
//					 rtp1.getPayload();
							 if (rtp1.getPayload().length == CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize()) {
//					 if(rtp1.getPayload().length>audioPacketSize-1) {
//					 System.arraycopy(rtp1.getPayload(),0,temp,copyPos,rtp1.getPayload().length);
//						 copyPos=copyPos+rtp1.getPayload().length;
								 System.arraycopy(rtp1.getPayload(), 0, temp, copyPos, CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize());
								 copyPos = copyPos + CaptureNetPacketService.getAppConfig().getRtpAudioPacketSize();
//						 log.error(" lose a little 少量丢包,正在补充ing ,当前位置补充{}，当前序号:{}, 正在补充: {} 随后序号: {} ", step - 1, currentSeq,currentSeq+k+1, nextSeq);
							 }
						 }
						 log.debug("packageRtpData: lose a little 少量丢包 ,当前位置需要补{}个包，当前序号:{}, 随后序号: {} ,--------> 补包后的位置: {} 临时数组大小: {}", step - 1, currentSeq, nextSeq, copyPos, bufferSize);
					 }
					 }
				 }
			 }catch (Exception e) {
				 log.error("packageRtpData: 复制数据出现错误，当前序号:{}, 随后序号: {} ,--------> 当前的位置: {} 临时数组大小: {} 有效负载长度: {} ExceptionMessage: {}",currentSeq,nextSeq,copyPos,bufferSize,rtp1.getPayload().length,e);

			}finally {
				 // 清理掉前面的垃圾数据
				 removeOldSeq(rtps,currentSeq);
				 //移除数据
				 rtps.remove(currentSeq);
			 }
		 }
			 log.debug("packageRtpDataSaveCache : 整理 后剩余: {} 个数据. 是否强制写入临时文件: {}",rtps.size(),forceWriteCache);


		 // 当前处理后的数组
		 factData=new byte[copyPos];
		    System.arraycopy(temp,0,factData,0,copyPos);
		    
		    //存储到临时临时缓存或缓存文件中
		 if(CaptureNetPacketService.getAppConfig().isSaveFile()) {
			if(fileCacheBytes==null) {
				fileCacheBytes=new byte[copyPos];
			  System.arraycopy(factData,0,fileCacheBytes,0,factData.length);
			}
			else fileCacheBytes=Tools.byteMerger(fileCacheBytes, factData);


			// 缓存设置值时，将文件缓存写入文件中。   需要将所有包写入缓存文件。
			if(fileCacheBytes.length>CaptureNetPacketService.getAppConfig().getSaveFileCacheMaxSize() || forceWriteCache) {
			// 写入缓存文件
//			WavFileUtil.appendToFile(temp, copyPos, pathtemp);

//			WavFileUtil.appendToFile(WavFileUtil.convertG711ToPcm(fileCacheBytes, null), fileCacheBytes.length, pathtemp);
			WavFileUtil.appendToFile(fileCacheBytes, fileCacheBytes.length, pathtemp);
			fileCacheBytes=null;
				log.debug("packageRtpDataSaveCache : 整理 后剩余: {} 个数据. {} 已写入文件: {}",rtps.size(),forceWriteCache,pathtemp);
			}
		 }
			 
		} catch (Exception e) {
		log.error("packageRtpData Error   rtp1.getPayload().length:{},temp.length: {}, copyPos:{}",rtp1.getPayload().length,temp.length,copyPos,e);
		}
		log.debug("本轮结束。packageRtpDataSaveCache : 整理 后剩余: {} 个数据. {} 已写入文件: {}",rtps.size(),forceWriteCache,pathtemp);
		return factData;
	}

	
	/**
	 * 	移除掉已经添加的seq，到当前包
	 */
	public void removeOldSeq(ConcurrentHashMap<Integer ,RtpDto> rtps,int currentSeq) {
		// 移除2倍批处理之前的数据到现在的seq
		int startRemoveSeq=currentSeq-2*CaptureNetPacketService.getAppConfig().getRtpBatchPackagesSave()+1 ;
		if(startRemoveSeq<0)startRemoveSeq=1;
		for (int i = startRemoveSeq ; i < currentSeq; i++) {
			if(rtps.containsKey(i))
			rtps.remove(i);
		}
	}
	
	
	private void creatWavFileInfo(long tempSimple, short audioFormat) throws UnsupportedEncodingException {
//		https://blog.csdn.net/u012478275/article/details/99623110
//		因此RTP的时间戳又称为媒体时间戳，以强调这种时间戳的粒度取决于信号的类型。例如， 对于8kHz采样的话音信号，若每隔20ms构成一个数据块，则一个数据块中包含有160个样本（0.02×8000=160）。因此每发送一个RTP分 组，其时间戳的值就增加160。
//		1、时间戳就是一个值，用来反映某个数据块的产生（采集）时间点的， 后采集的数据块的时间戳肯定是大于先采集的数据块的。有了这样一个时间戳，就可以标记数据块的先后顺序。
//		    2，在实时流传输中，数据采集后立刻传递到RTP 模块进行发送，那么，其实，数据块的采集时间戳就直接作为RTP包的时间戳。
//		    3，如果用RTP来传输固定的文件，则这个时间戳 就是读文件的时间点，依次递增。这个不再我们当前的讨论范围内，暂时不考虑。
//		    4，时间戳的单位采用的是采样频率的倒数，例如采 样频率为8000Hz时，时间戳的单位为1 / 8000 ，在Jrtplib库中，有设置时间戳单位的函数接口，而ORTP库中根据负载类型直接给定了时间戳的单位（音频负载1/8000，视频负载1 /90000）
//		   5，时间戳增量是指两个RTP包之间的时间间隔， 详细点说，就是发送第二个RTP包相距发送第一个RTP包时的时间间隔（单位是时间戳单位）。
		//所以下面的音频的采样率写死8000即可
		log.debug("creatWavFileInfo: tempSimple{} audioFormat:{}",tempSimple,audioFormat);
		// 音频6或者7.
		if(audioFormat==6 || audioFormat==7) {
//		  byte[] fact_data  = ByteBuffer.allocate(4).put("fact".getBytes("ISO_8859_1")).array();
//		  audioFormat=1;
			wavFileInfo=new WavFileInfo(audioFormat, (short)2, (int)8000, (short)8, 0);
//			wavFileInfo=new WavFileInfo(audioFormat, (short)2, (int)8000, (short)8, 0,(short)12,fact_data);
			}
		
	}

 

	 

	@Override
	public void analysisPackageData() {
		try {
			AnalysisRTPData();
			while((positiveRtps.size()>2*CaptureNetPacketService.getAppConfig().getRtpBatchPackagesSave() ||reverseRtps.size()>2*CaptureNetPacketService.getAppConfig().getRtpBatchPackagesSave() ))
			{
				log.info(" Rtp 数量大于 2倍 {} 需要加急处理一次。",CaptureNetPacketService.getAppConfig().getRtpBatchPackagesSave());
			AnalysisRTPData();
			Thread.sleep(20);
			isTimeOut=Tools.isTimeOutRtpPacket(lastGetPacketDateTime);
			}
			isTimeOut=Tools.isTimeOutRtpPacket(lastGetPacketDateTime);
			// 超时或者线程停止都需要处理完数据
				if(isTimeOut) {
					stopThread=true; 
				}
		 } catch (Exception e) {
			 log.error("startAnalysis packageRtpData Error {}",e);
		 }
	}

	@Override
	public void initAnalysis() {
		log.debug("starting  正在初始化 。initAnalysis IP: {}  positiveNetworkInfo： {}", positiveNetworkInfo.getDestinationIP(),Tools.getjsonString(positiveNetworkInfo));
		if(channelCount.getAndIncrement()>CaptureNetPacketService.getMaxChannelCount()) {
			log.warn("告警已经超过最大通道数 ，授权最大通道数{}",CaptureNetPacketService.getMaxChannelCount());
		};
                            iniRtpInfoThread=new Thread(()->{
							  initRtpInfo();
						  }
						  );
		iniRtpInfoThread.setPriority(NORM_PRIORITY-2);
		iniRtpInfoThread.setName("AnalysisRTP-firstSearchSip"+positiveNetworkInfo.getNetworkInfoKey());
		CaptureNetPacketService.getTaskService().execute(iniRtpInfoThread);
		log.debug("初始化已发起 。initAnalysis IP: {}  positiveNetworkInfo： {}", positiveNetworkInfo.getDestinationIP(),Tools.getjsonString(positiveNetworkInfo));
	}


	public void initRtpInfo(){
		log.debug("使用新线程 正在初始化Rtp信息 。iniRtpInfo IP: {}  positiveNetworkInfo： {}", positiveNetworkInfo.getDestinationIP(),Tools.getjsonString(positiveNetworkInfo));
		LocalDateTime currentTime=LocalDateTime.now();
		String tempFileString=currentTime.format(CaptureNetPacketService.tempFileDF);
		String tempDirString=CaptureNetPacketService.getAppConfig().getSaveFilePath()+File.separator+currentTime.format(CaptureNetPacketService.dirDF);

//		 String fileName=currentTime.format(CaptureNetPacketService.fileDF);
		String saveFileName= positiveNetworkInfo.getSourcePort()+"to"+ positiveNetworkInfo.getDestinationPort()+"t"+tempFileString+"egoo.wav";
		if(positivePath ==null) {
			positivePath =CaptureNetPacketService.getAppConfig().getTempFilePath()+File.separator+ positiveNetworkInfo.getSourcePort()+"positivePath"+ positiveNetworkInfo.getDestinationPort()+"t"+tempFileString+".temp";
			reversePath=CaptureNetPacketService.getAppConfig().getTempFilePath()+File.separator+ positiveNetworkInfo.getDestinationPort()+"reversePath"+ positiveNetworkInfo.getSourcePort()+"t"+tempFileString+".temp";
		}
		if(saveFilePath==null) {
			File dir=new File(tempDirString);
			if(!dir.exists())dir.mkdir();
			saveFilePath=tempDirString+File.separator+saveFileName;
		}
		//			第一个网络信息的IP的源地址不是服务器，说明客户端发起的流是第一个.服务器为接受方
		if(CaptureNetPacketService.getAppConfig().getRtpServerIPs().contains(positiveNetworkInfo.getDestinationIP())) {
			clientToServerIsInPositiveDirection =true;
			log.debug("客户端发送给服务器包在正向的positive中-----IP: {} 在服务器列表中。----------------positiveNetworkInfo>>>>>>{}", positiveNetworkInfo.getDestinationIP(),Tools.getjsonString(positiveNetworkInfo));
		} else {
			log.debug("服务器发送给客户端包在正向的positive中------IP: {} 不在服务器列表中。---------------positiveNetworkInfo>>>>>>{}", positiveNetworkInfo.getDestinationIP(),Tools.getjsonString(positiveNetworkInfo));

		}

		// 根据network信息获取到最后一次通话的信令数据，并将录音地址保存进去。
		// 信令有可能sip或者其他协议生成，录音地址取当前对象的
		if(signallingInfo==null)
			signallingInfo=waitSipSignallingInfo();

		if(signallingInfo!=null) {
			handelClientIP(signallingInfo);
			signallingInfo.setRecordPath(saveFilePath);
			// 开启新线程任务去解析信令和发送数据流给其他平台,分机号先知道，可以先推给登录分机的坐席
			try {
				if(RedisConfig.isLeader()) {
					if(deviceInfo!=null && deviceInfo.isNprtd()) {

						voiceInfoPush=new VoiceInfoPush(signallingInfo);
						voiceInfoPush.setDeviceInfo(deviceInfo);
						voiceInfoPush.setName("voiceInfoPush-Main"+positiveNetworkInfo.getNetworkInfoKey()+signallingInfo.getDeviceNumber());
						CaptureNetPacketService.getTaskService().execute(voiceInfoPush);
					}
				}
			}catch (Exception e) {
				log.error(" Exception 这条录音关联到信令  saveFilePath: {}。 \r\n positivePath: {}   reversePath: {} .......>>\r\n signallingInfo  {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo));
			}
			//获取信令
			if(CaptureNetPacketService.getAppConfig().isDependThirdParty() ) {
				try {
					log.debug(" 已有基础信令，正在获取依赖第三方获取更多信息。 signallingInfo 为。 {}",Tools.getjsonString(signallingInfo));
					CaptureNetPacketService.getDependThirdPartyService().improveAndPerfectSignallingInfo(signallingInfo);
					log.debug(" 已有基础信令。这条录音关联到信令，已经补充了信息。补充后 signallingInfo 为。 {}",Tools.getjsonString(signallingInfo));
				}catch (Exception e){
					log.warn(" 已经获取到信令，可能SIP或其他方式获取到。第三方信令扩展信息时，获取信令失败。  saveFilePath: {}。 \r\n positivePath: {}   reversePath: {} .......>>\r\n signallingInfo  {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo));
				}
			}
			log.debug(" 这条录音关联到信令  saveFilePath: {}。  positivePath: {}   reversePath: {} .......>> signallingInfo  {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo));
		}else if(signallingInfo==null &&CaptureNetPacketService.getAppConfig().isDependThirdParty()){
			//分机与IP对应关系并依赖第三方的信令的处理
			signallingInfo=new SignallingInfo();
			clientSignallingCallInfos.put(positiveNetworkInfo.getSourceIP()+">"+ positiveNetworkInfo.getSourcePort(),signallingInfo);
//				设置IP、mac与分机对应关系。
			handelClientIP(signallingInfo);
			try {
				if(RedisConfig.isLeader()) {
					// 开启新线程任务去解析信令和发送数据流给其他平台,分机号先知道，可以先推给登录分机的坐席
					if(deviceInfo!=null && deviceInfo.isNprtd()) {
						voiceInfoPush=new VoiceInfoPush(signallingInfo);
						voiceInfoPush.setDeviceInfo(deviceInfo);
						CaptureNetPacketService.getTaskService().execute(voiceInfoPush);
					}
				}
			}catch (Exception e) {
				log.error("Exception 这条录音关联到信令 由第三方提供  saveFilePath: {}。 \r\n positivePath: {}   reversePath: {} .......>>\r\n signallingInfo  {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo));
			}
			try{
				log.info(" handelClientIP 已经进行了IP与分机的对应，正在根据分机查找信令。signallingInfo:{}",signallingInfo);
				CaptureNetPacketService.getDependThirdPartyService().getSignallingInfo(signallingInfo);
			}catch (Exception e){
				log.warn(" 获取第三方信令失败。 这条录音关联到信令  saveFilePath: {}。 \r\n positivePath: {}   reversePath: {} .......>>\r\n signallingInfo  {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo));
			}
			log.debug(" 这条录音关联到信令 由第三方提供  saveFilePath: {}。 \r\n positivePath: {}   reversePath: {} .......>>\r\n signallingInfo  {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo));
		} else {
			log.debug(" 这条录音没有关联到信令,保存的录音地址为  saveFilePath: {}。 \r\n positivePath: {}   reversePath: {} .......>>\r\n signallingInfo  {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo));
		}

//			 if(signallingInfo.getRecordPath()==null)
		signallingInfo.setRecordPath(saveFilePath);
		log.debug("初始化Rtp信息完成 。iniRtpInfo IP: {}  positiveNetworkInfo： {}  信令: {}", positiveNetworkInfo.getDestinationIP(),Tools.getjsonString(positiveNetworkInfo),Tools.getjsonString(signallingInfo));
		iniRtpInfoThreadStoped=true;
	}

	/**
	 * 获取sip里面的信令基础信息，多台服务器时，sip信令生成的信令可能会晚于rtp时间。
	 * @return signallingInfo 信令
	 */
	public SignallingInfo waitSipSignallingInfo(){
		Object obj=new Object();
		long timeOut=CaptureNetPacketService.getAppConfig().getWaitSipSignallingInfoTimeOut();
		SignallingInfo signallingInfo=null;
		log.debug("waitSipSignallingInfo    开始获信令. timeOut:{}      当前流媒体网络信息: {}  根据网络信息获取。",timeOut, Tools.getjsonString(positiveNetworkInfo));
		while(signallingInfo==null && timeOut>0) {
			try {
				signallingInfo= getLastSignallingInfo(positiveNetworkInfo);
				if(signallingInfo!=null) {
					log.info("waitSipSignallingInfo  已经获取到信令: {}  剩余获取时间 {} 毫秒",timeOut,Tools.getjsonString(signallingInfo),timeOut);
					return signallingInfo;
				}
				synchronized (obj) {
					obj.wait(100);
				}
				timeOut=timeOut-100;
//				log.warn("waitSipSignallingInfo 剩余获取时间 {} 毫秒.positiveNetworkInfo : {}  ",timeOut,Tools.getjsonString(positiveNetworkInfo));
			} catch (Exception e) {
				log.error("waitSipSignallingInfo positiveNetworkInfo :{}  ,timeOut: {}",Tools.getjsonString(positiveNetworkInfo),timeOut,e);
			}
		}
		log.debug("waitSipSignallingInfo    positiveNetworkInfo{} 未获取到sip信令. timeOut:{}      网络信息: {}",timeOut,Tools.getjsonString(positiveNetworkInfo));
		return signallingInfo;
	}





	/**
	 * 
		* <strong>Title: <strong>
		* <p>Description: 根据IP信息处理ClientIP信息<p>
		* <p>Company: </p> 
		*<strong>Copyright:</strong> Copyright (c) 2019
		* @version: 1.0
		* @author  ASUS:augustWei
		* @date  2020 上午12:20:45
		* @param signInfo
	 */
	public void handelClientIP(SignallingInfo signInfo) {
		  
		 try {
		 //设置信令客户端的IP和mac
		 log.debug("handelClientIP--->>>>{},signInfo:{} \r\n positiveNetworkInfo:{}", clientToServerIsInPositiveDirection,Tools.getjsonString(signInfo),Tools.getjsonString(positiveNetworkInfo));
		  deviceInfo= CaptureNetPacketService.getStationDeviceInfos().get(positiveNetworkInfo.getSourceIP());
		 if(deviceInfo==null)
			 deviceInfo=CaptureNetPacketService.getStationDeviceInfos().get(positiveNetworkInfo.getSourceMac());
		 if(deviceInfo==null)
			 deviceInfo= CaptureNetPacketService.getStationDeviceInfos().get(positiveNetworkInfo.getDestinationIP());
		 if(deviceInfo==null)
			 deviceInfo= CaptureNetPacketService.getStationDeviceInfos().get(positiveNetworkInfo.getDestinationMac());
		 // 是否为其他类型的通道
		 if(deviceInfo==null)
			 deviceInfo= CaptureNetPacketService.getOtherDeviceInfos().get(positiveNetworkInfo.getSourceIP());
		 if(deviceInfo==null)
			 deviceInfo= CaptureNetPacketService.getOtherDeviceInfos().get(positiveNetworkInfo.getSourceMac());
		 if(deviceInfo==null)
			 deviceInfo= CaptureNetPacketService.getOtherDeviceInfos().get(positiveNetworkInfo.getDestinationIP());
		 if(deviceInfo==null)
			 deviceInfo= CaptureNetPacketService.getOtherDeviceInfos().get(positiveNetworkInfo.getDestinationMac());
		 
		 if(deviceInfo!=null) {
		 signInfo.setClientIP(positiveNetworkInfo.getSourceIP());
		 signInfo.setClientMAC(positiveNetworkInfo.getSourceMac());
		 signInfo.setServerIP(positiveNetworkInfo.getDestinationIP());
		 	}
		 else 
		 {
		  signInfo.setClientIP(positiveNetworkInfo.getDestinationIP());
		  signInfo.setClientMAC(positiveNetworkInfo.getDestinationMac());
		  signInfo.setServerIP(positiveNetworkInfo.getSourceIP());
		 }
		 log.debug("handelClientIP---<<<<<<<<<<<<<{}   end,signInfo: {}   positiveNetworkInfo:{}  获取到的deviceInfo {}", clientToServerIsInPositiveDirection,Tools.getjsonString(signInfo),Tools.getjsonString(positiveNetworkInfo),Tools.getjsonString(deviceInfo));
		 // 根据客户端MAC和IP查找设备号码
		  if(deviceInfo==null) deviceInfo=CaptureNetPacketService.getDeviceInfos().get(signInfo.getClientIP());
		 if(deviceInfo==null)deviceInfo=CaptureNetPacketService.getDeviceInfos().get(signInfo.getClientMAC());
		 if(deviceInfo!=null)
		 signInfo.setDeviceNumber(deviceInfo.getDeviceNumber());
		 
		 
		 }catch (Exception e) {
			 log.error("handelClientIP---<<<<<<<<<<<<<{}   end,signInfo: {}  \\r\\n positiveNetworkInfo:{}", clientToServerIsInPositiveDirection,Tools.getjsonString(signInfo),Tools.getjsonString(positiveNetworkInfo),e);
		}
	}
	
	@Override
	public void endAnalysis()  {
		try {
		// 超时的时候处理的
			channelCount.getAndDecrement();
			log.info("通道数使用量 channelCount :{}",channelCount.get());

			//		保留为0
			surplus=0;

			// 结束掉子线程。
			if(iniRtpInfoThread!=null){
				iniRtpInfoThread.interrupt();
                       while (!iniRtpInfoThreadStoped){
						   iniRtpInfoThread.interrupt();
					   }
					   this.join();
			}

			if(voiceInfoPush!=null){
				// 结束的时候被唤醒,并且关闭语音流推送的服务线程
				voiceInfoPush.setStopThread(true);
				voiceInfoPush.interrupt();
				this.join();
			}

			analysisSortPackageData();

			handelAllPackageData();

			if(CaptureNetPacketService.getAppConfig().isSaveFile() ) {
			 if(positiveReciveLastRtpDto ==null && reverseReciveLastRtpDto ==null) {
				 
				 log.info(" 这通语音没有数据流-  saveFilePath: {}。 \r\n positivePath: {}   reversePath: {} .......>>\r\n signallingInfo  {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo));
				 
		}else {
			if(deviceInfo!=null && deviceInfo.isNeedRecord())
			{
			saveToDb();
				log.info(" 已写数据库-  saveFilePath: {}。 \r\n positivePath: {}   reversePath: {} .......>>\r\n signallingInfo  {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo));
			}else {
				log.info(" 未写数据库请检查分机录音是否配置。-  saveFilePath: {}。 \r\n positivePath: {}   reversePath: {} .......>>\r\n signallingInfo  {} deviceInfo: {}",saveFilePath, positivePath,reversePath,Tools.getjsonString(signallingInfo),Tools.getjsonString(deviceInfo));
			}
				 Path positivePathObj= Paths.get(positivePath);
				 Path reversePathObj= Paths.get(reversePath);
			if(Files.exists(positivePathObj) && Files.exists(reversePathObj) && Files.size(positivePathObj)>0 && Files.size(reversePathObj)>0 ) {
				//如果客户端发送的第一个包，坐席声音在第一声道。始终保证客户端发起的包是第一个。
				log.info("写入双声道文件，positivePath: {}   大小: {}  和 reversePath: {}  大小: {}。", positivePath,Files.size(positivePathObj),reversePath,Files.size(reversePathObj));
				if (!clientToServerIsInPositiveDirection) {
					// 服务器发给坐席的，为客户的声音
					WavFileUtil.mergeWave(wavFileInfo, positivePath, reversePath, saveFilePath);

				} else {
					WavFileUtil.mergeWave(wavFileInfo, reversePath, positivePath, saveFilePath);
				}
				log.info("写入双声道文件完成，positivePath: {}  大小: {}和 reversePath: {} 不存在。", positivePath,reversePath);
			}else if(Files.exists(positivePathObj) && Files.size(positivePathObj)>0){
				log.info("第一声道写入单道文件，positivePath: {} 大小: {} ", positivePath,Files.size(positivePathObj));
				WavFileUtil.mergeWave(wavFileInfo, positivePath, null, saveFilePath);
				log.info("第一声道写入单道文件完成，positivePath: {} 大小: {} ", positivePath,Files.size(positivePathObj));
			} else if(Files.exists(reversePathObj) && Files.size(reversePathObj)>0){
				log.info("第二声道写入单道文件，reversePath: {} 大小: {} ",reversePath,Files.size(reversePathObj));
				WavFileUtil.mergeWave(wavFileInfo, null, reversePath, saveFilePath);
				log.info("第二声道写入单道文件完成，reversePath: {} 大小: {} ",reversePath,Files.size(reversePathObj));
			}else {
				log.info("positivePath: {}  和 reversePath: {} 不存在。", positivePath,reversePath);
			}



//			wavFileInfo=new WavFileInfo((short)6, (short)1, (int)8000, (short)8, 0);
//			WavFileUtil.mergeWave(wavFileInfo, PositivePath,null, saveFilePath.replacePositive("egoo", "Positive_egoo"));
//			WavFileUtil.mergeWave(wavFileInfo, reversePath,null,  saveFilePath.replacePositive("egoo", "reverse__egoo"));
		}
			 
		}
		 
		}catch (Exception e) {
			log.error("endAnalysis: {}  Exception",signallingInfo,e);
		}finally {

			// 清理掉信令
			analysisRTPs.remove(positiveNetworkInfo.getNetworkInfoKey());
			analysisRTPs.remove(reverseNetworkInfo.getNetworkInfoKey());
			if(signallingInfo!=null) {
				signallingInfo.clear(positiveNetworkInfo);
				signallingInfo=null;
			}
			positiveReciveLastRtpDto=null;
			reverseReciveLastRtpDto=null;
			log.debug("flow endAnalysis 处理录音结束 。iniRtpInfo IP: {}  positiveNetworkInfo： {}  信令: {}", positiveNetworkInfo.getDestinationIP(),Tools.getjsonString(positiveNetworkInfo),Tools.getjsonString(signallingInfo));
			/*positiveRtps.clear();
			reverseRtps.clear();*/
			positiveRtpsTemp.clear();
			positiveRtpsTemp=null;
			reverseRtpsTemp.clear();
			reverseRtpsTemp=null;
			positiveRtps.clear();
			positiveRtps=null;
			reverseRtps.clear();
			reverseRtps=null;
			wavFileInfo=null;
			deviceInfo=null;
			positiveFileCacheBytes=null;
			reverseFileCacheBytes=null;
			initTime=null;
			iniRtpInfoThread=null;
		}
	}

	@Override
	public void handelAllPackageData() {
		 

		log.info(" handelAllPackageRtpData {} rtpsSize:{} reverseRtpsSize:{} !!! ", positiveNetworkInfo.getNetworkInfoKey(), positiveRtps.size(),reverseRtps.size());
		packageRtpDataSaveCache(positiveRtps, positivePath, positiveFileCacheBytes,false);
		packageRtpDataSaveCache(reverseRtps, reversePath, reverseFileCacheBytes,false);
		packageRtpDataSaveCache(positiveRtps, positivePath, positiveFileCacheBytes,true);
		packageRtpDataSaveCache(reverseRtps,reversePath,reverseFileCacheBytes,true);
	}
	 
	 public void saveToDb(){
		 try {
			 RecordLog recordLog=new RecordLog();
			 recordLog.setId(UUID.randomUUID()+"");
			 if(signallingInfo==null)
				 signallingInfo= waitSipSignallingInfo();
			 if(signallingInfo.getCallId()!=null) 
			 recordLog.setCallId(signallingInfo.getCallId());
			 if(signallingInfo.getCallUUID()!=null) 
			 recordLog.setCallUUID(signallingInfo.getCallUUID());
			 if(signallingInfo.getAgentID()!=null) 
				 recordLog.setAgentID(signallingInfo.getAgentID());
			 if(signallingInfo.getAppData()!=null) {
			 recordLog.setAppDataMap(signallingInfo.getAppData());
			 recordLog.setAppData(Tools.getjsonString((recordLog.getAppDataMap())));
			 }
			 if(signallingInfo.getFromUser()!=null) {
			 recordLog.setFromUser(signallingInfo.getFromUser());
			 recordLog.setCaller(signallingInfo.getCaller());
			 }
			 if(signallingInfo.getToUser()!=null) {
			 recordLog.setToUser(signallingInfo.getToUser());
			 recordLog.setCalled(signallingInfo.getCalled());
			 }
			if(deviceInfo!=null)
			 if(deviceInfo.getDeviceNumber()!=null)
			 recordLog.setDeviceNumber(deviceInfo.getDeviceNumber());
			 
			 if(signallingInfo.getClientIP()!=null)
				 recordLog.setClientIP(signallingInfo.getClientIP());
			 if(signallingInfo.getServerIP()!=null)
			 recordLog.setServerIP(signallingInfo.getServerIP());
			 if(signallingInfo.getEndCase()!=null)
			 recordLog.setEndCase(signallingInfo.getEndCase());
			 
			 recordLog.setDirection(signallingInfo.getDirection());
			 
			 // 信令只记第一次的路径，所以不能取信令的
//			 if(signallingInfo.getRecordPath()!=null)
			 recordLog.setRecordPath(saveFilePath);
			 
			 
			 signallingInfo.setEstablishTime(startTime);
			 signallingInfo.setEndTime(lastGetPacketDateTime);
			 
			 if(signallingInfo.getBeginTime()!=null) {
			 recordLog.setBeginTime(signallingInfo.getBeginTime().format(Tools.dfFull));
			 recordLog.setBeginTimestamp(signallingInfo.getBeginTime().toInstant(ZoneOffset.of("+8")).toEpochMilli());
			 }
			 
			 // 振铃事件为空设置开始时间
			 if(signallingInfo.getRingTime()==null) {
				 signallingInfo.setRingTime(signallingInfo.getBeginTime());
			 }
			 
			 
			 
			 if(signallingInfo.getRingTime()!=null)
			 {
			 recordLog.setRingTime(signallingInfo.getRingTime().format(Tools.dfFull));
			 recordLog.setRingTimestamp(signallingInfo.getRingTime().toInstant(ZoneOffset.of("+8")).toEpochMilli());
			 }
			 
			 recordLog.setEstablishTime(signallingInfo.getEstablishTime().format(Tools.dfFull));
			 recordLog.setEstablishTimestamp(signallingInfo.getEstablishTime().toInstant(ZoneOffset.of("+8")).toEpochMilli());
			 
			 
			 
			 
			  
			 
			 recordLog.setEndTime(signallingInfo.getEndTime().format(Tools.dfFull));
			 recordLog.setEndTimestamp(signallingInfo.getEndTime().toInstant(ZoneOffset.of("+8")).toEpochMilli());
			 
			 
			 recordLog.setClusterServerId(CaptureNetPacketService.getClusterServId());

			 if(RedisConfig.isLeader()) {
				 RedisConfig.publishRecQueueMessage("recLog", recordLog);
				 RedisConfig.publishMessage("recLog",recordLog);
				 log.debug("当前节点为主节点，已发布录音信息。recordLog 录音记录  : {}   \r\n signallingInfo: {}",Tools.getjsonString(recordLog),Tools.getjsonString(signallingInfo));
			 }else {
				 log.debug("当前节点为非节点，无需发布录音信息。recordLog 录音记录  : {}   \r\n signallingInfo: {}",Tools.getjsonString(recordLog),Tools.getjsonString(signallingInfo));
			 }
		 }catch (Exception e) {
			 log.error("saveToDb signallingInfo : {} ,networkInfo: {}:",e,Tools.getjsonString(positiveNetworkInfo));
		}
	 }
      
	 
	 public static void garbageCollection(){
		 analysisRTPs.entrySet().forEach(analysisRTP->{
					 analysisRTP.getValue().awaken();
				 }
		 );
	 }
}
